package com.example.toy.core.writer;

import com.example.toy.core.commit.CommitMsg;
import com.example.toy.core.dataset.BigCollection;
import com.example.toy.core.fs.ToyTableVersionFileView;
import com.example.toy.core.merge.MergeMode;
import com.example.toy.core.record.ToyRecord;
import com.example.toy.core.record.ToyRecordImpl;
import com.example.toy.core.record.key.ToyRecordKey;
import com.example.toy.core.table.ToyTableMeta;
import com.example.toy.core.version.ToyVersion;

import java.util.List;
import java.util.Optional;

public class DefaultToyTableWriter implements ToyTableWriter {

  private final ToyTableMeta tableMeta;

  public DefaultToyTableWriter(ToyTableMeta tableMeta) {
    this.tableMeta = tableMeta;
  }

  /**
   * @param records
   * @param mode
   * @return
   */
  @Override
  public BigCollection<ToyRecordKey> save(
      BigCollection<ToyRecord> records, WriteScope scope, MergeMode mode, HardMode hardMode) {

    ToyVersion toyVersion = ToyVersion.nextVersion();


    ToyTableVersionFileView fileView = tableMeta.getVersionFileView();

    BigCollection<ToyRecordKey> results =
        records
            .groupBy(rec -> rec.getKey().getPartitionKey())
            .map(
                pair -> {
                  String partitionKey = pair.getLeft();
                  Iterable<ToyRecord> partitionRecords = pair.getRight();

                  boolean usingBaseFileWriter =
                      scope != WriteScope.record || fileView.getBaseFiles(partitionKey).isEmpty();

                  ToyFileWriter writer =
                      usingBaseFileWriter
                          ? new DefaultBaseFileWriter(null,null)
                          : new DefaultDeltaFileWriter();
                  List<ToyRecordKey> wroteKeys = writer.write(partitionRecords);
                  return wroteKeys;
                })
            .flatMap(keys -> keys.iterator());

    return results;
  }

  @Override
  public BigCollection<ToyRecordKey> update(BigCollection<ToyRecord> records) {
    return save(records, WriteScope.record, MergeMode.update, HardMode.soft);
  }

  @Override
  public long deleteByKey(BigCollection<ToyRecordKey> recordKeys, HardMode hardMode) {
    BigCollection<ToyRecord> records = recordKeys.map(key -> new ToyRecordImpl());
    return delete(records, hardMode);
  }

  @Override
  public long delete(BigCollection<ToyRecord> records, HardMode hardMode) {
    save(records, WriteScope.record, MergeMode.delete, hardMode);
    return 0;
  }

  public CommitMsg startCommit(WriteScope scope, MergeMode mode, HardMode hardMode) {
    ToyVersion toyVersion = ToyVersion.nextVersion();
    return null;
  }

  public void finishCommit(CommitMsg msg, Optional<Exception> exceptionOptional) {

  }

  @Override
  public void startTransaction() {}

  @Override
  public void commitTransaction() {}

  @Override
  public void rollbackTransaction() {}
}
